package com.exer.dataload;


import com.exer.dataload.config.DriverConfig;
import com.exer.dataload.log.DorisDataLoadLog;
import com.exer.dataload.utils.Utils;
import org.apache.http.HttpHeaders;
import org.apache.http.client.methods.CloseableHttpResponse;
import org.apache.http.client.methods.HttpPut;
import org.apache.http.entity.StringEntity;
import org.apache.http.impl.client.CloseableHttpClient;
import org.apache.http.impl.client.DefaultRedirectStrategy;
import org.apache.http.impl.client.HttpClientBuilder;
import org.apache.http.impl.client.HttpClients;
import org.apache.http.util.EntityUtils;

import java.sql.*;
import java.util.*;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 将mysql数据导入到doris中
 */
public class DorisDataLoaderFromMySQL implements DorisDataLoad {


    private static final ThreadPoolExecutor THREAD_POOL;

    private static final HttpClientBuilder httpClientBuilder = HttpClients
            .custom()
            .setRedirectStrategy(new DefaultRedirectStrategy() {
                @Override
                protected boolean isRedirectable(String method) {
                    return true;
                }
            });
    private String tableName;

    public DorisDataLoaderFromMySQL(String tableName) {
        this.tableName = tableName;
    }

    static {
        //初始化线程池
        int i = Runtime.getRuntime().availableProcessors();
        int corePoolSize = i;
        int maximumPoolSize = i * 2;
        int BlockingQueueSize = 1024;
        THREAD_POOL = new ThreadPoolExecutor(corePoolSize, maximumPoolSize, 4L, TimeUnit.MINUTES, new LinkedBlockingQueue<>(BlockingQueueSize), new ThreadPoolExecutor.DiscardPolicy());
    }


    /**
     * 检查doris中表是否存在
     *
     * @return true存在，false不存在
     */
    @Override
    public boolean dorisTableExistenceChecker() {
        try (Connection dorisConn = DriverConfig.getDorisConnection();
             Statement stmt = dorisConn.createStatement();
             ResultSet rs = stmt.executeQuery("SHOW TABLES LIKE '" + this.tableName + "'")) {
            if (rs.next()) {
                return true;
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
        return false;
    }

    /**
     * 在doris中创建表
     *
     * @param dorisDDL doris建表语句
     * @return 是否创建成功
     */
    @Override
    public boolean createDorisTable(String dorisDDL) {
        try (Connection dorisConn = DriverConfig.getDorisConnection();
             Statement stmt = dorisConn.createStatement()) {
            return stmt.execute(dorisDDL);
        } catch (SQLException e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 检查doris表中是否有数据
     *
     * @return 0则无数据
     */
    @Override
    public long getDorisTableCount() {
        try (Connection dorisConn = DriverConfig.getDorisConnection();
             Statement stmt = dorisConn.createStatement();
             ResultSet resultSet = stmt.executeQuery("select id from " + this.tableName + " limit 1")) {
            int count = 0;
            while (resultSet.next()) {
                count++;
            }
            return count;
        } catch (SQLException e) {
            e.printStackTrace();
            return 0;
        }
    }

    /**
     * 构建streamLoad导入的数据
     *
     * @param resultSet mysql查询结果集
     * @return streamLoad导入的数据
     */
    private DataLoadEntity buildStreamLoadContent(ResultSet resultSet) {
        if (resultSet == null) return null;
        try {
            StringBuilder content = new StringBuilder();
            long id = -1;
            while (resultSet.next()) {
                for (int i = 1; i <= resultSet.getMetaData().getColumnCount(); i++) {
                    String data = resultSet.getString(i);
                    content.append(data).append("\t");
                }
                content.replace(content.length() - 1, content.length(), "\n");
                id = resultSet.getLong("id");
            }
            //防止resultSet没有数据
            if (content.length() != 0) {
                content.deleteCharAt(content.length() - 1);
                return new DataLoadEntity(content.toString(), id);
            }
        } catch (SQLException e) {
            e.printStackTrace();
        } finally {
            try {
                resultSet.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
        return null;
    }

    /**
     * 用于增量同步时查询mysql的数据
     *
     * @param dorisLastUpdateTime 最后更新时间
     * @param flag                停止标识
     * @param queue               存放数据的阻塞队列
     * @param dataImportStrategy  数据导入策略
     */
    private void readMysqlData(Object dorisLastUpdateTime, AtomicBoolean flag, BlockingQueue<DataLoadEntity> queue, IDataImportStrategy<ResultSet, DataLoadEntity> dataImportStrategy) {
        try (Connection conn = DriverConfig.getMysqlConnection();
             PreparedStatement stmt = conn.prepareStatement(
                     "SELECT t1.* FROM " + this.tableName + " t1," +
                             "(SELECT t.id FROM " + this.tableName + " t WHERE t.update_time > ? ORDER BY t.update_time LIMIT ?,?)" +
                             " t2 WHERE t1.id = t2.id")) {
            int offset = 0;
            int limit = 2000;
            stmt.setObject(1, dorisLastUpdateTime);
            stmt.setObject(3, limit);
            while (flag.get()) {
                long startTime = System.currentTimeMillis();
                stmt.setLong(2, offset);
                ResultSet resultSet = stmt.executeQuery();
                DataLoadEntity DataLoadEntity = dataImportStrategy.getDate(resultSet);
                //返回null表示查不出数据,设置停止标识
                if (DataLoadEntity == null) {
                    flag.set(false);
                    break;
                }
                queue.put(DataLoadEntity);
                offset += limit;
                long endTime = System.currentTimeMillis();
                DorisDataLoadLog.log(DorisDataLoadLog.LogType.INFO,
                        "DorisDataLoaderFromMySQL",
                        "readMysqlData",
                        "从mysql读取" + limit + "条数据完成,耗时" + DorisDataLoadLog.formatDuration(endTime - startTime));
            }
        } catch (SQLException | InterruptedException e) {
            e.printStackTrace();
        }
    }

    /**
     * 用于全量同步时查询mysql的数据
     *
     * @param flag               停止标识
     * @param queue              存放数据的阻塞队列
     * @param dataImportStrategy 数据导入策略
     */
    private void readMysqlData(AtomicBoolean flag, BlockingQueue<DataLoadEntity> queue, IDataImportStrategy<ResultSet, DataLoadEntity> dataImportStrategy) {
        try (Connection conn = DriverConfig.getMysqlConnection();
             PreparedStatement stmt = conn.prepareStatement(
                     "SELECT * FROM " + this.tableName + " where id > ? limit ?")) {
            //设置起始分页id
            long id = -1;
            int limit = 2000;
            stmt.setObject(2, limit);
            while (flag.get()) {
                long startTime = System.currentTimeMillis();
                stmt.setLong(1, id);
                ResultSet resultSet = stmt.executeQuery();
                DataLoadEntity DataLoadEntity = dataImportStrategy.getDate(resultSet);
                //返回null表示查不出数据,设置停止标识
                if (DataLoadEntity == null) {
                    flag.set(false);
                    break;
                }
                //重新设置起始分页id
                id = DataLoadEntity.lastId;
                queue.put(DataLoadEntity);
                long endTime = System.currentTimeMillis();
                DorisDataLoadLog.log(DorisDataLoadLog.LogType.INFO,
                        "DorisDataLoaderFromMySQL",
                        "readMysqlData",
                        "从mysql读取" + limit + "条数据完成,耗时" + DorisDataLoadLog.formatDuration(endTime - startTime));
            }
        } catch (SQLException | InterruptedException e) {
            e.printStackTrace();
        }
    }


    /**
     * 导入数据到doris，以InsertInto语句的方式，进行增量同步
     * doris官方不推荐
     */
    @Override
    public void dorisDataLoadByInsertIntoIncremental() {
        try (Connection dorisConnection = DriverConfig.getDorisConnection();
             Statement statement = dorisConnection.createStatement();
             ResultSet dorisResultSet = statement.executeQuery("select update_time from " + this.tableName + " order by update_time desc limit 1")) {
            if (!dorisResultSet.next()) {
                throw new RuntimeException("Doris " + this.tableName + " 表中无数据! 请使用全量同步！");
            }
            //获取doris最后一个更新时间
            Object dorisLastUpdateTime = dorisResultSet.getObject(1);
            //创建结束标识,数据队列
            AtomicBoolean flag = new AtomicBoolean(true);
            BlockingQueue<DataLoadEntity> queue = new LinkedBlockingQueue<>();
            //读线程,生产者
            DorisDataLoadLog.submitWithLogging(THREAD_POOL, "DorisDataLoaderFromMySQL", "readMysqlData",
                    "\n[-----查询结束-----]",
                    () -> readMysqlData(dorisLastUpdateTime, flag, queue, this::buildDorisInsertSQL));
            //写线程,消费者
            DorisDataLoadLog.submitWithLogging(THREAD_POOL, "DorisDataLoaderFromMySQL", "executeDorisInsert",
                    "\n[-----本次同步结束-----]",
                    () -> executeDorisInsert(flag, queue));
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    /**
     * 根据表名找出id差异集合,并以streamLoad的方式插入到doris中
     */
    @Override
    public void dorisDataLoadByIdDifferences() {
        //创建结束标识,数据队列
        AtomicBoolean flag = new AtomicBoolean(true);
        BlockingQueue<Set<Long>> queue = new LinkedBlockingQueue<>();
        DorisDataLoadLog.submitWithLogging(THREAD_POOL, "DorisDataLoaderFromMySQL", "findIdDifferences",
                "\n[-----查找差异ID结束-----]",
                () -> findIdDifferences(flag, queue));
        DorisDataLoadLog.submitWithLogging(THREAD_POOL, "DorisDataLoaderFromMySQL", "dorisDataLoadByIdDifferences",
                "\n[-----本次同步结束-----]",
                () -> dorisDataLoadByIdDifferences(flag, queue));
    }


    /**
     * 将id差异集回表查询后以streamLoad的方式插入到doris
     *
     * @param flag  停止标识
     * @param queue 数据队列
     */
    private void dorisDataLoadByIdDifferences(AtomicBoolean flag, BlockingQueue<Set<Long>> queue) {
        //创建结束标识,数据队列
        AtomicBoolean stopFlag = new AtomicBoolean(true);
        BlockingQueue<DataLoadEntity> dataLoadEntityQueue = new LinkedBlockingQueue<>();
        //根据id差异集回表查询
        THREAD_POOL.submit(() -> {
            try (Connection conn = DriverConfig.getMysqlConnection()) {
                while (flag.get() || !queue.isEmpty()) {
                    Set<Long> idSet = queue.poll();
                    DataLoadEntity dataLoadEntity = readMysqlData(conn, idSet);
                    if (dataLoadEntity == null) continue;
                    dataLoadEntityQueue.put(dataLoadEntity);
                }
            } catch (SQLException | InterruptedException e) {
                e.printStackTrace();
            }
            stopFlag.set(false);
        });
        //将回表查询的数据插入到doris
        while (stopFlag.get() || !dataLoadEntityQueue.isEmpty()) {
            sendData(stopFlag, dataLoadEntityQueue);
        }
    }

    /**
     * 根据id差异集回表查询
     *
     * @param conn  mysql连接
     * @param idSet id差异集
     * @return dorisStreamLoad导入对象
     */
    private DataLoadEntity readMysqlData(Connection conn, Set<Long> idSet) {
        if (idSet == null || idSet.size() == 0) {
            return null;
        }
        //构建mysql回表查询语句
        StringBuilder sqlBuilder = new StringBuilder("select * from ").append(this.tableName).append(" where id in (");
        for (int i = 0; i < idSet.size(); i++) {
            sqlBuilder.append("? ").append(",");
        }
        sqlBuilder.deleteCharAt(sqlBuilder.length() - 1);
        sqlBuilder.append(")");
        try (PreparedStatement stmt = conn.prepareStatement(sqlBuilder.toString())) {
            int count = 1;
            Set<Long> set = Collections.synchronizedSet(idSet);
            for (Long id : set) {
                stmt.setObject(count, id);
                count++;
            }
            ResultSet resultSet = stmt.executeQuery();
            return buildStreamLoadContent(resultSet);
        } catch (SQLException e) {
            e.printStackTrace();
        }
        return null;
    }

    /**
     * 查询id差异集
     *
     * @param flag  停止标识
     * @param queue 数据队列
     */
    private void findIdDifferences(AtomicBoolean flag, BlockingQueue<Set<Long>> queue) {
        try (Connection dorisConn = DriverConfig.getDorisConnection();
             Connection mysqlConn = DriverConfig.getMysqlConnection()) {
            //设置起始id作为分页条件
            AtomicLong mysqlStartId = new AtomicLong(-2);
            AtomicLong dorisStartId = new AtomicLong(-2);
            String sql = "select id from " + this.tableName + " where id > ? order by id limit 500000";
            //当mysql和doris中查询无数据时,起始id设置为-1,停止循环
            while (mysqlStartId.get() != -1 || dorisStartId.get() != -1) {
                long sTime = System.currentTimeMillis();
                //mysql查询线程,返回mysql的idSet集合用于找出差异id
                CompletableFuture<Set<Long>> mysqlFuture = CompletableFuture.supplyAsync(() -> {
                    if (mysqlStartId.get() == -1) return null;
                    try (PreparedStatement mysqlStmt = mysqlConn.prepareStatement(sql)) {
                        mysqlStmt.setLong(1, mysqlStartId.get());
                        try (ResultSet mysqlResultSet = mysqlStmt.executeQuery()) {
                            long mysqlLastId = -1;
                            Set<Long> idSet = new HashSet<>();
                            while (mysqlResultSet.next()) {
                                mysqlLastId = mysqlResultSet.getLong(1);
                                idSet.add(mysqlLastId);
                            }
                            mysqlStartId.set(mysqlLastId);
                            return idSet;
                        }
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                    return null;
                }, THREAD_POOL);
                //doris查询线程,返回doris的idSet集合用于找出差异id
                CompletableFuture<Set<Long>> dorisFuture = CompletableFuture.supplyAsync(() -> {
                    if (dorisStartId.get() == -1) return null;
                    try (PreparedStatement dorisStmt = dorisConn.prepareStatement(sql)) {
                        dorisStmt.setLong(1, dorisStartId.get());
                        try (ResultSet dorisResultSet = dorisStmt.executeQuery()) {
                            long dorisLastId = -1;
                            Set<Long> idSet = new HashSet<>();
                            while (dorisResultSet.next()) {
                                dorisLastId = dorisResultSet.getLong(1);
                                idSet.add(dorisLastId);
                            }
                            dorisStartId.set(dorisLastId);
                            dorisResultSet.close();
                            return idSet;
                        }
                    } catch (SQLException e) {
                        e.printStackTrace();
                    }
                    return null;
                }, THREAD_POOL);
                // 汇总两个id集合计算差集线程,返回最终差异id集合
                Set<Long> set = mysqlFuture.thenCombineAsync(dorisFuture, (idSetA, idSetB) -> {
                    if (idSetA == null && idSetB == null) return null;
                    if (idSetA == null) {
                        return idSetB;
                    }
                    if (idSetB == null) {
                        return idSetA;
                    }
                    //计算对称差集
                    Set<Long> diff1to2 = new HashSet<>(idSetA);
                    diff1to2.removeAll(idSetB);
                    Set<Long> diff2to1 = new HashSet<>(idSetB);
                    diff2to1.removeAll(idSetA);
                    Set<Long> symmetricDiff = new HashSet<>(diff1to2);
                    symmetricDiff.addAll(diff2to1);
                    return symmetricDiff;
                }).join();
                if (set != null)
                    //由于doris插入限制,将id差异集进行分批
                    createBatches(set, queue, 4000);
                long eTime = System.currentTimeMillis();
                DorisDataLoadLog.log(DorisDataLoadLog.LogType.INFO, "计算差异id完成,耗时:" + DorisDataLoadLog.formatDuration(eTime - sTime));
            }
        } catch (SQLException | InterruptedException e) {
            e.printStackTrace();
        }
        //设置停止标识
        flag.set(false);
    }

    /**
     * 根据batchSize将set集合的数据分批后存放到队列中
     *
     * @param set       要分批的set集合
     * @param queue     数据队列
     * @param batchSize 每个批次的大小
     */
    private void createBatches(Set<Long> set, BlockingQueue<Set<Long>> queue, int batchSize) throws InterruptedException {
        if (queue == null || set == null || batchSize <= 1) return;
        long totalSize = set.size();
        if (totalSize == 0) return;
        if (totalSize <= batchSize) {
            queue.put(set);
            return;
        }
        long i = 0;
        Set<Long> resultSet = new HashSet<>();
        for (Long id : set) {
            resultSet.add(id);
            i++;
            if (i == batchSize) {
                queue.put(resultSet);
                resultSet = new HashSet<>();
                i = 0;
            }

        }
        if (!resultSet.isEmpty()) {
            queue.put(resultSet);
        }
    }


    /**
     * 构建doris插入对象,会关闭ResultSet!!!
     *
     * @param resultSet mysql查询返回的结果集
     * @return doris插入对象
     */
    private DataLoadEntity buildDorisInsertSQL(ResultSet resultSet) {
        if (resultSet == null) return null;
        try {
            int columnCount = resultSet.getMetaData().getColumnCount();
            StringBuilder queryBuilder = new StringBuilder();
            //构建sql语句
            queryBuilder.append("INSERT INTO ").append(this.tableName).append(" VALUES ");
            List<List<Object>> resultList = new ArrayList<>();
            long lastId = -1;
            while (resultSet.next()) {
                queryBuilder.append("(");
                List<Object> result = new ArrayList<>();
                for (int i = 1; i <= columnCount; i++) {
                    result.add(resultSet.getObject(i));
                    queryBuilder.append("? ").append(",");
                }
                resultList.add(result);
                queryBuilder.setLength(queryBuilder.length() - 1);
                queryBuilder.append("),");
                //获取本次插入的最后一个id用于分页处理
                lastId = resultSet.getLong("id");
            }

            queryBuilder.setLength(queryBuilder.length() - 1);
            queryBuilder.append(";");
            //当结果集无数据时返回null,表示查不出任何数据
            if (resultList.size() == 0) {
                return null;
            }
            return new DataLoadEntity(queryBuilder.toString(), resultList, lastId);
        } catch (SQLException e) {
            e.printStackTrace();
            return null;
        }finally {
            try {
                resultSet.close();
            } catch (SQLException e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 导入数据到doris，以InsertInto语句的方式，进行全量同步
     * doris官方不推荐
     */
    @Override
    public void dorisDataLoadByInsertIntoFull() {
        //创建结束标识,数据队列
        AtomicBoolean flag = new AtomicBoolean(true);
        BlockingQueue<DataLoadEntity> queue = new LinkedBlockingQueue<>();
        //读线程,生产者
        DorisDataLoadLog.submitWithLogging(THREAD_POOL, "DorisDataLoaderFromMySQL", "readMysqlData",
                "\n[-----查询结束-----]",
                () -> readMysqlData(flag, queue, this::buildDorisInsertSQL));
        //写线程,消费者
        DorisDataLoadLog.submitWithLogging(THREAD_POOL, "DorisDataLoaderFromMySQL", "executeDorisInsert",
                "\n[-----本次同步结束-----]",
                () -> executeDorisInsert(flag, queue));
    }


    /**
     * 导入数据到doris，以StreamLoad的方式，进行全量同步
     */
    @Override
    public void dorisDataLoadByStreamLoadFull() {
        //创建结束标识,数据队列
        AtomicBoolean flag = new AtomicBoolean(true);
        BlockingQueue<DataLoadEntity> queue = new LinkedBlockingQueue<>();
        //读线程,生产者
        DorisDataLoadLog.submitWithLogging(THREAD_POOL, "DorisDataLoaderFromMySQL", "readMysqlData",
                "\n[-----查询结束-----]",
                () -> readMysqlData(flag, queue, this::buildStreamLoadContent));
        //写线程,消费者
        DorisDataLoadLog.submitWithLogging(THREAD_POOL, "DorisDataLoaderFromMySQL", "sendData",
                "\n[-----本次同步结束-----]",
                () -> sendData(flag, queue));
    }

    /**
     * 导入数据到doris，以StreamLoad的方式，进行增量同步
     */
    @Override
    public void dorisDataLoadByStreamLoadIncremental() {
        try (Connection dorisConnection = DriverConfig.getDorisConnection();
             Statement statement = dorisConnection.createStatement();
             ResultSet dorisResultSet = statement.executeQuery("select update_time from " + this.tableName + " order by update_time desc limit 1")) {
            if (!dorisResultSet.next()) {
                throw new RuntimeException("Doris " + this.tableName + " 表中无数据! 请使用全量同步！");
            }
            //获取doris最后一个更新时间
            Object dorisLastUpdateTime = dorisResultSet.getObject(1);
            //创建结束标识,数据队列
            AtomicBoolean flag = new AtomicBoolean(true);
            BlockingQueue<DataLoadEntity> queue = new LinkedBlockingQueue<>();
            //读线程,生产者
            DorisDataLoadLog.submitWithLogging(THREAD_POOL, "DorisDataLoaderFromMySQL", "readMysqlData",
                    "\n[-----查询结束-----]",
                    () -> readMysqlData(dorisLastUpdateTime, flag, queue, this::buildStreamLoadContent));
            //写线程,消费者
            DorisDataLoadLog.submitWithLogging(THREAD_POOL, "DorisDataLoaderFromMySQL", "sendData",
                    "\n[-----本次同步结束-----]",
                    () -> sendData(flag, queue));
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }

    /**
     * 以streamLoad的方式将数据发送到doris
     *
     * @param flag  结束标识
     * @param queue 存放数据的队列
     */
    private void sendData(AtomicBoolean flag, BlockingQueue<DataLoadEntity> queue) {
        try (CloseableHttpClient client = httpClientBuilder.build()) {
            //准备http请求信息
            HttpPut put = new HttpPut(DriverConfig.getDorisStreamLoadUrl(this.tableName));
            //设置label后续可检查本次导入的信息
            //put.setHeader("label", "39c25a5c-7000-496e-a98e-348a264c81de");
            put.setHeader(HttpHeaders.EXPECT, "100-continue");
            put.setHeader(HttpHeaders.AUTHORIZATION, DriverConfig.basicAuthHeader());
            //当结束标识和队列为空时停止
            while (flag.get() || !queue.isEmpty()) {
                //从队列中取得数据
                DataLoadEntity dataLoadEntity = queue.poll();
                if (dataLoadEntity == null) continue;
                StringEntity entity = new StringEntity(dataLoadEntity.dorisStreamLoadContent, "UTF-8");
                put.setEntity(entity);
                //发送请求并处理响应
                try (CloseableHttpResponse response = client.execute(put)) {
                    String loadResult = "";
                    if (response.getEntity() != null) {
                        loadResult = EntityUtils.toString(response.getEntity());
                    }
                    int statusCode = response.getStatusLine().getStatusCode();
                    if (statusCode != 200) {
                        DorisDataLoadLog.log(DorisDataLoadLog.LogType.ERROR,
                                "Stream load failed, statusCode=" + statusCode + " load result=" + loadResult);
                    }
                    //打印日志
                    loadResult = DorisDataLoadLog.formatStreamLoadResponse(loadResult);
                    String status = Utils.findSub(loadResult, "Status\":\\s*\"?([^\",]+)\"?,", 1);
                    if ("Success".equals(status)){
                        DorisDataLoadLog.log(DorisDataLoadLog.LogType.INFO,
                                "DorisDataLoaderFromMySQL",
                                "sendData",
                                loadResult);
                    }else {
                        DorisDataLoadLog.log(DorisDataLoadLog.LogType.WARN,
                                "DorisDataLoaderFromMySQL",
                                "sendData",
                                loadResult);
                    }
                }
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }



    /**
     * 将数据以InsertINTO的方式插入到doris,
     * doris官方不推荐
     *
     * @param flag  停止标识符
     * @param queue 数据队列
     */
    private void executeDorisInsert(AtomicBoolean flag, BlockingQueue<DataLoadEntity> queue) {
        try (Connection conn = DriverConfig.getDorisConnection()) {
            while (flag.get() || !queue.isEmpty()) {
                //从队列中取得数据
                DataLoadEntity dataLoadEntity = queue.poll();
                if (dataLoadEntity == null) continue;
                long sTime = System.currentTimeMillis();
                //将数据填充进statement,并发送请求
                try (PreparedStatement statement = conn.prepareStatement(dataLoadEntity.dorisInsertSql)) {
                    int count = 1;
                    for (List<Object> objects : dataLoadEntity.resultList) {
                        for (int i = 0; i < objects.size(); i++) {
                            statement.setObject(count, objects.get(i));
                            count++;
                        }
                    }
                    statement.executeUpdate();
                }
                long eTime = System.currentTimeMillis();
                DorisDataLoadLog.log(DorisDataLoadLog.LogType.INFO, "执行doris插入完成,耗时:" + DorisDataLoadLog.formatDuration(eTime - sTime));
            }
        } catch (SQLException e) {
            e.printStackTrace();
        }
    }


    /**
     * 私有内部类
     * doris插入对象类,
     * 属性包含插入语句和数据和最后的id用于分页
     */
    private static class DataLoadEntity {
        /**
         * StreamLoad方式导入的内容
         */
        private String dorisStreamLoadContent;
        /**
         * doris插入语句
         */
        private String dorisInsertSql;
        /**
         * mysql查询结果数据
         */
        private List<List<Object>> resultList;
        /**
         * 本次查询最后一条的id
         */
        private long lastId;

        public DataLoadEntity(String dorisStreamLoadContent, long lastId) {
            this.dorisStreamLoadContent = dorisStreamLoadContent;
            this.lastId = lastId;
        }

        public DataLoadEntity(String dorisInsertSql, List<List<Object>> resultList, long lastId) {
            this.dorisInsertSql = dorisInsertSql;
            this.resultList = resultList;
            this.lastId = lastId;
        }
    }

    @FunctionalInterface
    private interface IDataImportStrategy<T, R> {
        R getDate(T t);
    }

}
